/*-------------------------------------------------------------------------
 *
 * hive_funcs.c
 *                Function Mapping handler for Hive
 *
 * This file contains logic to identify which built-in functions can be
 * sent to Hive. These functions are part of the WHERE clause. In some
 * cases we also need to translate the function name into something
 * which is understood and accepted by Hive
 *
 * Copyright (c) 2012-2013, BigSQL Development Group
 *
 * IDENTIFICATION
 *                hadoop_fdw/src/hive_funcs.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "hadoop_fdw.h"

#include "access/htup_details.h"
#include "catalog/pg_proc.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"

/*
 * We need to track the set of functions which can be sent to the remote
 * side. Additionally some of these functions will have another name on
 * the remote side. We track this in an array of simple 'hive_funcs_mapping'
 * structure. This array is stored in a sorted manner to allow efficient
 * lookup at run time via the use of bsearch
 *
 * For polymorphic functions, we can track the number of arguments required
 * by them. For example log with 1 argument is equivalent to "log10" on the
 * Hive side and log with 2 arguments is equivalent to "log"
 *
 * For other functions like round, the name is the same on both sides
 * regardless of the number of arguments.
 *
 * The time[stamp] related functions in Hive are very
 * different from the ones in PG as well.
 *
 * PG allows dates/timestamps in different formats. Hive
 * assumes a fixed format for most functions.
 *
 * We can convert date_part() calls into equivalent
 * year, month, day, hour, minute, second calls
 *
 * To handle most of the above cases we use a
 * special Hive funcname "hive_translate". The code when it encounters
 * this will add specific code to translate into Hive understandable
 * function names.
 *
 * List of functions on the Hive side that are not supported by PostgreSQL
 *
 * log2, bin, unhex, conv, positive, negative, e, from_utc_timestamp,
 * to_utc_timestamp, find_in_set, format_number, get_json_object,
 * in_file, locate, ngrams, parse_url, printf, regexp_extract,
 * regexp_replace, sentences, space, split, str_to_map, percentile,
 * percentile_approx, histogram_number, collect_set, inline,
 */
struct hive_func_mapping
{
	const char *fname;		/* name of the function in PostgreSQL */
	const char *hive_fname; /* mapping of the function on Hive side */
};

/*
 * Generated by referring to the following link:
 *
 * https://cwiki.apache.org/Hive/languagemanual-udf.html
 *
 * NOTE: Always add new entries in SORT order!
 */
static struct hive_func_mapping HiveFunctionMap[] =
{
	{ "abs",			"abs"					},
	{ "acos",			"acos"					},
	{ "ascii",			"ascii"					},
	{ "asin",			"asin"					},
	{ "atan",			"atan"					},
	{ "avg",			"avg"					},
	{ "ceil",			"ceil"					},
	{ "ceiling",		"ceiling"				},
	{ "concat",			"concat"				},
	{ "concat_ws",		"concat_ws"				},
	{ "corr",			"corr"					},
	{ "cos",			"cos"					},
	{ "count",			"count"					},
	{ "covar_pop",		"covar_pop"				},
	{ "covar_samp",		"covar_samp"			},
	{ "date_mii",		"date_sub"				},
	{ "date_part",		"hive_translate"		}, /* convert into specific Hive functions */
	{ "date_pli",		"date_add"				},
	{ "degrees",		"degrees"				},
	{ "exp",			"exp"					},
	{ "floor",			"floor"					},
	{ "length",			"hive_translate"		}, /* only 1 arg format supported in Hive */
	{ "ln",				"ln"					},
	{ "log",			"hive_translate"		}, /* convert into specific Hive functions */
	{ "lower",			"lower"					},
	{ "lpad",			"hive_translate"		}, /* only 3 args format supported in Hive */
	{ "ltrim",			"hive_translate"		}, /* only 1 arg format supported in Hive */
	{ "max",			"max"					},
	{ "min",			"min"					},
	{ "mod",			"pmod"					}, /* does it work ok with only +ve values? */
	{ "pi",				"pi"					},
	{ "power",			"power"					},
	{ "radians",		"radians"				},
	{ "random",			"rand"					},
	{ "repeat",			"repeat"				},
	{ "reverse",		"reverse"				},
	{ "round",			"round"					},
	{ "rpad",			"hive_translate"		}, /* only 1 arg format supported in Hive */
	{ "rtrim",			"hive_translate"		}, /* only 1 arg format supported in Hive */
	{ "sign",			"sign"					},
	{ "sin",			"sin"					},
	{ "sqrt",			"sqrt"					},
	{ "stddev_pop",		"stddev_pop"			},
	{ "stddev_samp",	"stddev_samp"			},
	{ "strpos",			"instr"					},
	{ "substr",			"substr"				},
	{ "sum",			"sum"					},
	{ "tan",			"tan"					},
	{ "to_hex",			"hex"					},
	{ "to_timestamp",	"hive_translate"		},
	{ "translate",		"translate"				},
	{ "trim",			"hive_translate"		}, /* only 1 arg format supported in Hive */
	{ "unnest",			"explode"				},
	{ "upper",			"upper"					},
	{ "variance",		"variance"				},
	{ "var_pop",		"var_pop"				},
	{ "var_samp",		"var_samp"				}
};

/*
 * TODO: Check if xpath functions need to be sent to Hive as well
 */

static int	hive_func_compare(const void *a, const void *b);
static int	hive_func_name_compare(const char *namea, const char *nameb);


/*
 * Provided a function expression and the name of that function, use
 * the HiveFunctionMap structure to come up with a proper name. In many
 * cases the translation is straightforward.
 *
 * However in some cases, the hive fname "hive_translate" means that due
 * to polymorphism some additional translations will have to be carried
 * out
 */
const char *
hive_translate_function(FuncExpr *fe, const char *fname)
{
	struct			hive_func_mapping	key;
	struct			hive_func_mapping  *res;
	int				num_functions;

	num_functions = sizeof(HiveFunctionMap)/
						sizeof(struct hive_func_mapping);

	key.fname = fname;
	key.hive_fname = NULL;
	res = (struct hive_func_mapping *) bsearch(&key,
											 HiveFunctionMap,
											 num_functions,
											 sizeof(struct hive_func_mapping),
											 hive_func_compare);

	/* This should never happen.. */
	if (res == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_FDW_ERROR),
				 errmsg("unable to map function: %s", fname)
				));

	elog(DEBUG2, "translated %s to %s", fname, res->hive_fname);
	if (strcmp(res->hive_fname, "hive_translate") != 0)
		return res->hive_fname;


	/* Single argument is log base 10 */
	if (strcmp(fname, "log") == 0)
	{
		if (list_length(fe->args) == 1)
			return "log10";
		else
			return "log";
	}

	/* to_timestamp stuff */
	if (strcmp(fname, "to_timestamp") == 0)
	{
		if (list_length(fe->args) == 1)
			return "from_unixtime";
		else
			return "unix_timestamp";
	}

	/* date_part stuff */
	if (strcmp(fname, "date_part") == 0)
	{
		/* get the first argument */
		Expr *expr = linitial(fe->args);
		if (nodeTag(expr) == T_Const)
		{
			Const *node = (Const *)expr;
			Oid			typoutput;
			bool		typIsVarlena;
			char	   *quantum;

			getTypeOutputInfo(node->consttype,
							  &typoutput, &typIsVarlena);
			quantum = OidOutputFunctionCall(typoutput, node->constvalue);

			if (strcmp(quantum, "year") == 0)
				return "year";

			if (strcmp(quantum, "month") == 0)
				return "month";

			if (strcmp(quantum, "day") == 0)
				return "day";

			if (strcmp(quantum, "hour") == 0)
				return "hour";

			if (strcmp(quantum, "minute") == 0)
				return "minute";

			if (strcmp(quantum, "second") == 0)
				return "second";

			if (strcmp(quantum, "week") == 0)
				return "weekofyear";
		}
	}

	/* return ltrim, rtrim, trim, length, lpad, rpad as is */
	return fname;
}

/*
 * Check if the incoming function can be sent to Hive. We consult the
 * HiveFunctionMap table for this. For performance we use bsearch to
 * quickly find if the function is shippable or not. This is just the
 * lookup function.
 */
bool
is_hive_builtin(FuncExpr *fe)
{
	struct			hive_func_mapping	key;
	struct			hive_func_mapping  *res;
	int				num_functions;
	HeapTuple		proctup;
	Form_pg_proc	procform;
	const char	   *proname;

	proctup = SearchSysCache1(PROCOID, ObjectIdGetDatum(fe->funcid));
	if (!HeapTupleIsValid(proctup))
		elog(ERROR, "cache lookup failed for function %u", fe->funcid);
	procform = (Form_pg_proc) GETSTRUCT(proctup);

	/* Deparse the function name ... */
	proname = NameStr(procform->proname);
	ReleaseSysCache(proctup);

	num_functions = sizeof(HiveFunctionMap)/
						sizeof(struct hive_func_mapping);

	key.fname = proname;
	key.hive_fname = NULL;
	res = (struct hive_func_mapping *) bsearch(&key,
											 HiveFunctionMap,
											 num_functions,
											 sizeof(struct hive_func_mapping),
											 hive_func_compare);
	if (res == NULL)
		return false;

	/* if it's a standard mapping return immediately */
	if (strcmp(res->hive_fname, "hive_translate") != 0)
		return true;

	/* Only 1 argument ltrim function supported */
	if (strcmp(proname, "ltrim") == 0 && list_length(fe->args) != 1)
		return false;

	/* Only 1 argument rtrim function supported */
	if (strcmp(proname, "rtrim") == 0 && list_length(fe->args) != 1)
		return false;

	/* Only 1 argument trim function supported */
	if (strcmp(proname, "trim") == 0 && list_length(fe->args) != 1)
		return false;

	/* Only 1 argument length function supported */
	if (strcmp(proname, "length") == 0 && list_length(fe->args) != 1)
		return false;

	/*
	 * No need to waste cycles checking these:
	 * lpad, rpad, to_timestamp, date_part, log
	 */

	return true;
}

/*
 * Comparator for bsearching HiveFunctionMap array
 */
static int
hive_func_compare(const void *a, const void *b)
{
	const struct hive_func_mapping *confa =
					(const struct hive_func_mapping *) a;
	const struct hive_func_mapping *confb =
					(const struct hive_func_mapping *) b;

	return hive_func_name_compare(confa->fname, confb->fname);
}

/*
 * The bare comparison function for function names
 */
static int
hive_func_name_compare(const char *namea, const char *nameb)
{
	/*
	 * The temptation to use strcasecmp() here must be resisted, because the
	 * array ordering has to remain stable across setlocale() calls. So, build
	 * our own with a simple ASCII-only downcasing.
	 */
	while (*namea && *nameb)
	{
		char		cha = *namea++;
		char		chb = *nameb++;

		if (cha >= 'A' && cha <= 'Z')
			cha += 'a' - 'A';
		if (chb >= 'A' && chb <= 'Z')
			chb += 'a' - 'A';
		if (cha != chb)
			return cha - chb;
	}
	if (*namea)
		return 1;				/* a is longer */
	if (*nameb)
		return -1;				/* b is longer */
	return 0;
}
